package db.prob.operators.join;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import db.prob.io.LineParser;
import db.prob.io.PTuple;

public class ReducerJoin extends Reducer<Text, Text, Text, Text> {

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		if (key.toString().equals(MapperJoin.PASS)) {
			for (Text value : values) {
				context.write(null, value);
			}
			return;
		}
		
		Configuration conf = context.getConfiguration();
		List<PTuple> tuplesFromTable1 = new ArrayList<PTuple>();
		List<PTuple> tuplesFromTable2 = new ArrayList<PTuple>();
		
		populateTuplesLists(values, conf, tuplesFromTable1, tuplesFromTable2);
		// We don't need this attribute since it appears twice.
		int table2Attr = conf.getInt(MapperJoin.TABLE_2_ATTR, Integer.MAX_VALUE);
		String joinTableName = conf.get(MapperJoin.RESULT_TABLE);
		
		Text newVal = new Text();
		
		// Doing the join
		for (PTuple tuple1 : tuplesFromTable1) {
			for (PTuple tuple2 : tuplesFromTable2) {
				float newProb = tuple1.getProbability() * tuple2.getProbability();
				String[] newAttr = new String[tuple1.getAttr().length + tuple2.getAttr().length];
				System.arraycopy(tuple1.getAttr(), 0, newAttr, 0, tuple1.getAttr().length);
				// copy all except for the duplicated value.
				tuple2.getAttr()[table2Attr] = "";
				System.arraycopy(tuple2.getAttr(), 0, newAttr, tuple1.getAttr().length, tuple2.getAttr().length);
				PTuple newTuple = new PTuple(joinTableName, newProb, newAttr);
				newVal.set(newTuple.toString());
				context.write(null, newVal);
			}
		}
		
	}

	/**
	 * @param values
	 * @param conf 
	 * @param tuplesFromTable1
	 * @param tuplesFromTable2
	 */
	private static String populateTuplesLists(Iterable<Text> values, Configuration conf, List<PTuple> tuplesFromTable1, List<PTuple> tuplesFromTable2) {
		
		String table1Name = conf.get(MapperJoin.TABLE_1);
		String table2Name = conf.get(MapperJoin.TABLE_2);
		
		Iterator<Text> iter = values.iterator();
		
		PTuple tuple;
		for (Text textTuple : values) {
			tuple = LineParser.createPTuple(textTuple.toString(), "\\t");
			if (tuple.getTableName().equals(table1Name)) {
				tuplesFromTable1.add(tuple);
			} else {
				table2Name = tuple.getTableName();
				tuplesFromTable2.add(tuple);
			}			
		}
				
		return table1Name + "$" + table2Name;
	}
	
	public static int size(Iterable<?> iter) {
		int count = 0;
		for (Object object : iter) {
			count++;
		}
		return count;
	}

}
